package com.matrix.product.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author 有点甜
 * @since 2025/8/4
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class CanalClient implements CommandLineRunner {

    @Value("${canal.client.server-host}")
    private String canalServerHost;

    @Value("${canal.client.server-port}")
    private int canalServerPort;

    @Value("${canal.client.destination}")
    private String destination;

    private final RocketMQTemplate rocketMQTemplate;

    @Override
    public void run(String... args) {
        new Thread(() -> {
            CanalConnector connector = CanalConnectors.newSingleConnector(
                    new InetSocketAddress(canalServerHost, canalServerPort), destination, "", "");
            try {
                connector.connect();
                connector.subscribe(".*\\..*");
                connector.rollback();
                log.info("Canal 客户端启动成功，开始监听数据库变更...");
                while (true) {
                    Message message = connector.getWithoutAck(100);
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    if (batchId == -1 || size == 0) {
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                        }
                    } else {
                        handleEntries(message.getEntries());
                    }
                    connector.ack(batchId);
                }
            } catch (Exception e) {
                log.error("Canal 客户端运行异常", e);
            } finally {
                connector.disconnect();
            }
        }).start();
    }

    private void handleEntries(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            CanalEntry.RowChange rowChange;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("解析 Canal Entry 失败", e);
            }

            String tableName = entry.getHeader().getTableName();
            if ("t_product".equals(tableName) || "t_product_category".equals(tableName)) {
                CanalEntry.EventType eventType = rowChange.getEventType();
                log.info("监听到表 {} 发生变更, 类型: {}", tableName, eventType);

                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    String shopId = getColumnValue(rowData.getAfterColumnsList(), "shop_id");
                    if (shopId != null) {
                        Map<String, String> payload = new HashMap<>();
                        payload.put("type", "shop_menu");
                        payload.put("key", shopId);
                        rocketMQTemplate.convertAndSend("CACHE_SYNC_TOPIC", payload);
                        log.info("发送缓存同步消息到MQ, shopId: {}", shopId);
                    }
                }
            }
        }
    }

    private String getColumnValue(List<CanalEntry.Column> columns, String columnName) {
        return columns.stream()
                .filter(c -> c.getName().equalsIgnoreCase(columnName))
                .map(CanalEntry.Column::getValue)
                .findFirst()
                .orElse(null);
    }
}
